package com.audaque.springboot.foshanupload.web.kafkademo.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.concurrent.ExecutionException;

/**
 * @author zgb
 * @desc ...
 * @date 2023-08-28 19:09:59
 */
@Slf4j
@RestController
@RequestMapping("/kafka/adminClient")
public class AdminClientController {

    @Autowired
    private AdminClient adminClient;

    /**
     * 创建主题并配置分区和副本数
     * @throws InterruptedException
     */
    @GetMapping("/createTopic")
    public void createTopic() throws InterruptedException {
        NewTopic topic = new NewTopic("testTopic9", 10, (short) 1);
        adminClient.createTopics(Arrays.asList(topic));
        Thread.sleep(1000);
    }

    @GetMapping("/selectTopicInfo")
    public void selectTopicInfo() throws ExecutionException, InterruptedException {
        DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList("testTopic8"));
        result.all().get().forEach((k,v)->System.out.println("k: "+k+" ,v: "+v.toString()+"\n"));
    }


}
